package cas.ihep.hmss;

import cas.ihep.hmss.proto.HmssProtos;
import cas.ihep.hmss.queue.JobQueue;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.io.*;

public class Server {
    public void bind(int port,final JobQueue queue) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) {
                    ChannelPipeline pipeline=ch.pipeline();
                    pipeline.addLast("frameEncoder",new ProtobufVarint32LengthFieldPrepender());
                    pipeline.addLast("encoder",new ProtobufEncoder());
                    pipeline.addLast("frameDecoder",new ProtobufVarint32FrameDecoder());
                    pipeline.addLast("decoder",new ProtobufDecoder(HmssProtos.JobSubmitRequest.getDefaultInstance()));
                    pipeline.addLast(new ServerHandler(queue));
                }
            });
            ChannelFuture f = b.bind(port).sync();

            System.out.println("init start");
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    static class TempFile implements Closeable{
        private File ifile;
        TempFile(String prefix,String suffix,File basedir)throws IOException{
            ifile=File.createTempFile(prefix,suffix,basedir);
        }
        String getAbsolutePath(){
            return ifile.getAbsolutePath();
        }
        File get(){
            return ifile;
        }
        public void close(){
            if (ifile!=null){
                ifile.delete();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Class<?> klass=Class.forName(args[0]);
        Class<?> interfaces[]=klass.getInterfaces();
        boolean implemented=false;
        for(Class<?> cls:interfaces){
            if(JobQueue.class.equals(cls)){
                implemented=true;
                break;
            }
        }
        if (!implemented){
            System.err.println(args[1]+" is not a sub-class of cas.ihep.hmss.queue.JobQueue");
            return;
        }
        int port = 8080;
        try {
            port = Integer.valueOf(args[1]);
        } catch (NumberFormatException ignored) {

        }
        Object obj=klass.newInstance();
        SubmitThread sThread = new SubmitThread((JobQueue)obj);
        sThread.start();
        //Thread sThread = new Thread(new subThread());
        new Server().bind(port,(JobQueue)obj);
        sThread.stop=true;
        sThread.join();
    }
}
